/*
 * Copyright (c) www.bugull.com
 */

package com.bugull.farm.core.server;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;

import com.bugull.farm.core.cache.PropertiesCache;
import com.bugull.mongo.BuguConnection;
import com.bugull.mongo.BuguFramework;
import com.bugull.redis.RedisConnection;
import com.bugull.redis.listener.TopicListener;
import com.bugull.redis.mq.MQClient;

import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import io.netty.channel.EventLoopGroup;

import org.apache.logging.log4j.*;

/**
 * Abstract Netty NIO Server.
 * 
 * @author Frank Wen(xbwen@hotmail.com)
 */
public abstract class FarmServer {
    
    protected final static Logger logger = LogManager.getLogger(FarmServer.class);
    
    protected EventLoopGroup bossGroup;
    protected EventLoopGroup workerGroup;
    protected RedisConnection redisConnection;
    protected BuguConnection mongoConnection;
    private static final int threadNum = Runtime.getRuntime().availableProcessors()*2+1;
    private static ExecutorService executors = Executors.newFixedThreadPool(threadNum);
    protected PropertiesCache cache = PropertiesCache.getInstance();
    private static ScheduledThreadPoolExecutor scheduledExecutors = (ScheduledThreadPoolExecutor)Executors
            .newScheduledThreadPool(5);
    
    /**
     * 根据配置文件中的默认连接信息连接mongoDB
     */
    protected void connectSingleMongodbByDefault() throws Exception{
        connectSingleMongodb(cache.get("mongo.host"), cache.getInt("mongo.port"), cache.get("mongo.db"),
        		cache.get("mongo.username"), cache.get("mongo.password"));
    }
    
    /**
     * 根据数据库连接信息建立mongoDB连接
     * @param host
     * @param port
     * @param db
     * @param username
     * @param password
     */
    protected void connectSingleMongodb(String host, int port, String db, String username, String password) throws Exception{
    	mongoConnection = BuguFramework.getInstance().createConnection();
        mongoConnection.connect(host, port, db, username, password);
    }

    /**
     * 根据配置文件中的默认信息连接mongodb副本集
     * @throws Exception
     */
    protected void connectReplicaSetMongodbByDefault() throws Exception{
        //create mongodb connection
        List<ServerAddress> serverList = new ArrayList<ServerAddress>();
        serverList.add(new ServerAddress(cache.get("mongo.host.primary"), cache.getInt("mongo.port")));
        serverList.add(new ServerAddress(cache.get("mongo.host.secondary"), cache.getInt("mongo.port")));
        serverList.add(new ServerAddress(cache.get("mongo.host.thirdly"), cache.getInt("mongo.port")));
        List<MongoCredential> credentialList = new ArrayList<MongoCredential>();
        MongoCredential credentialA = MongoCredential.createCredential(cache.get("mongo.username"),
                cache.get("mongo.db"), cache.get("mongo.password").toCharArray());
        MongoCredential credentialB = MongoCredential.createCredential(cache.get("mongo.username"),
                cache.get("mongo.db"), cache.get("mongo.password").toCharArray());
        MongoCredential credentialC = MongoCredential.createCredential(cache.get("mongo.username"),
                cache.get("mongo.db"), cache.get("mongo.password").toCharArray());
        credentialList.add(credentialA);
        credentialList.add(credentialB);
        credentialList.add(credentialC);
        BuguConnection conn = BuguFramework.getInstance().createConnection();
        conn.setServerList(serverList).setCredentialList(credentialList).setDatabase(cache.get("mongo.db")).connect();
    }
    
    /**
     * 项目中如果使用redis作为消息服务器，连接redis服务器并使用listener监听指定的topic
     * 未指定redis服务器地址和认证密码时，使用FarmConstant中的默认值
     * @param listener
     */
    protected void listenRedisTopic(String topic, TopicListener listener) throws Exception{
    	listenRedisTopic(cache.get("redis.host"), cache.getInt("redis.port"), cache.get("redis.password"), topic, listener);
    }
    
    /**
     * 项目中如果使用redis作为消息服务器，连接redis服务器并使用listener监听指定的topic
     * @param host
     * @param topic
     * @param password
     * @param listener
     */
    protected void listenRedisTopic(String host, int port, String password, String topic, TopicListener listener) throws Exception{
    	redisConnection = RedisConnection.getInstance();
        redisConnection.setHost(host);
        redisConnection.setPort(port);
        redisConnection.setPassword(password);
        redisConnection.connect();
        MQClient client = redisConnection.getMQClient();
        client.setTopicListener(listener);
        client.subscribe(topic);
    }
    
    protected void doShutdown(){
        try{
            if(bossGroup != null){
                bossGroup.shutdownGracefully().sync();
            }
            if(workerGroup != null){
                workerGroup.shutdownGracefully().sync();
            }
            if(redisConnection != null){
                redisConnection.disconnect();
            }
            if(mongoConnection != null){
                mongoConnection.close();
            }
            if(executors != null){
            	executors.shutdownNow();
            }
            if (scheduledExecutors != null) {
                scheduledExecutors.shutdown();
                logger.debug("关闭计划任务线程池!!!");
            }
            ZKClient.getInstance().disconnect();
            logger.warn("Server has been shutdown gracefully!");
        }catch(Exception ex){
            logger.error("Error when shutdown server", ex);
        } 
    }
    
    protected void addShutdownHook() {
        Runtime runtime = Runtime.getRuntime();
        runtime.addShutdownHook(new Thread(){
            @Override  
            public void run() {
                System.out.println("执行 addShutdownHook...");
               logger.error("执行 addShutdownHook...");
                doShutdown();
            } 
        });
    }

    public static ExecutorService getExecutors() {
		return executors;
	}

    public static ScheduledThreadPoolExecutor getScheduledExecutors() {
        return scheduledExecutors;
    }
}
